<!DOCTYPE html>
<html lang="zh-cn">
<head>
  <meta charset="utf-8">
  <meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1">
  <title>RxJava基本语法源码分析 - smartzheng</title>
  <meta name="renderer" content="webkit" />
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1"/>

<meta http-equiv="Cache-Control" content="no-transform" />
<meta http-equiv="Cache-Control" content="no-siteapp" />

<meta name="theme-color" content="#f8f5ec" />
<meta name="msapplication-navbutton-color" content="#f8f5ec">
<meta name="apple-mobile-web-app-capable" content="yes">
<meta name="apple-mobile-web-app-status-bar-style" content="#f8f5ec">


<meta name="author" content="smartzheng" /><meta name="description" content="最近看了下网上的RxJava源码分析，发现所基于的源码版本和最新的略有不同，于是自己动手翻阅了一下最新的源码版本（rxjava:2.2.8，" /><meta name="keywords" content="Hugo, theme, even" />






<meta name="generator" content="Hugo 0.68.3 with theme even" />


<link rel="canonical" href="https://smartzheng.github.io/post/rxjava%E5%9F%BA%E6%9C%AC%E8%AF%AD%E6%B3%95%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90/" />
<link rel="apple-touch-icon" sizes="180x180" href="/apple-touch-icon.png">
<link rel="icon" type="image/png" sizes="32x32" href="/favicon-32x32.png">
<link rel="icon" type="image/png" sizes="16x16" href="/favicon-16x16.png">
<link rel="manifest" href="/manifest.json">
<link rel="mask-icon" href="/safari-pinned-tab.svg" color="#5bbad5">


<link href="/dist/even.c2a46f00.min.css" rel="stylesheet">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/@fancyapps/fancybox@3.1.20/dist/jquery.fancybox.min.css" integrity="sha256-7TyXnr2YU040zfSP+rEcz29ggW4j56/ujTPwjMzyqFY=" crossorigin="anonymous">


<meta property="og:title" content="RxJava基本语法源码分析" />
<meta property="og:description" content="最近看了下网上的RxJava源码分析，发现所基于的源码版本和最新的略有不同，于是自己动手翻阅了一下最新的源码版本（rxjava:2.2.8，" />
<meta property="og:type" content="article" />
<meta property="og:url" content="https://smartzheng.github.io/post/rxjava%E5%9F%BA%E6%9C%AC%E8%AF%AD%E6%B3%95%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90/" />
<meta property="article:published_time" content="2019-05-05T00:00:00+00:00" />
<meta property="article:modified_time" content="2019-05-05T00:00:00+00:00" />
<meta itemprop="name" content="RxJava基本语法源码分析">
<meta itemprop="description" content="最近看了下网上的RxJava源码分析，发现所基于的源码版本和最新的略有不同，于是自己动手翻阅了一下最新的源码版本（rxjava:2.2.8，">
<meta itemprop="datePublished" content="2019-05-05T00:00:00&#43;00:00" />
<meta itemprop="dateModified" content="2019-05-05T00:00:00&#43;00:00" />
<meta itemprop="wordCount" content="2143">



<meta itemprop="keywords" content="" /><meta name="twitter:card" content="summary"/>
<meta name="twitter:title" content="RxJava基本语法源码分析"/>
<meta name="twitter:description" content="最近看了下网上的RxJava源码分析，发现所基于的源码版本和最新的略有不同，于是自己动手翻阅了一下最新的源码版本（rxjava:2.2.8，"/>

<!--[if lte IE 9]>
  <script src="https://cdnjs.cloudflare.com/ajax/libs/classlist/1.1.20170427/classList.min.js"></script>
<![endif]-->

<!--[if lt IE 9]>
  <script src="https://cdn.jsdelivr.net/npm/html5shiv@3.7.3/dist/html5shiv.min.js"></script>
  <script src="https://cdn.jsdelivr.net/npm/respond.js@1.4.2/dest/respond.min.js"></script>
<![endif]-->

</head>
<body>
  <div id="mobile-navbar" class="mobile-navbar">
  <div class="mobile-header-logo">
    <a href="/" class="logo">smartzheng</a>
  </div>
  <div class="mobile-navbar-icon">
    <span></span>
    <span></span>
    <span></span>
  </div>
</div>
<nav id="mobile-menu" class="mobile-menu slideout-menu">
  <ul class="mobile-menu-list">
    <a href="/">
        <li class="mobile-menu-item">Home</li>
      </a><a href="/post/">
        <li class="mobile-menu-item">Archives</li>
      </a><a href="/tags/">
        <li class="mobile-menu-item">Tags</li>
      </a><a href="/categories/">
        <li class="mobile-menu-item">Categories</li>
      </a>
  </ul>
</nav>
  <div class="container" id="mobile-panel">
    <header id="header" class="header">
        <div class="logo-wrapper">
  <a href="/" class="logo">smartzheng</a>
</div>

<nav class="site-navbar">
  <ul id="menu" class="menu">
    <li class="menu-item">
        <a class="menu-item-link" href="/">Home</a>
      </li><li class="menu-item">
        <a class="menu-item-link" href="/post/">Archives</a>
      </li><li class="menu-item">
        <a class="menu-item-link" href="/tags/">Tags</a>
      </li><li class="menu-item">
        <a class="menu-item-link" href="/categories/">Categories</a>
      </li>
  </ul>
</nav>
    </header>

    <main id="main" class="main">
      <div class="content-wrapper">
        <div id="content" class="content">
          <article class="post">
    
    <header class="post-header">
      <h1 class="post-title">RxJava基本语法源码分析</h1>

      <div class="post-meta">
        <span class="post-time"> 2019-05-05 </span>
        <div class="post-category">
            <a href="/categories/android/"> Android </a>
            </div>
        
      </div>
    </header>

    <div class="post-toc" id="post-toc">
  <h2 class="post-toc-title">文章目录</h2>
  <div class="post-toc-content always-active">
    <nav id="TableOfContents"></nav>
  </div>
</div>
    <div class="post-content">
      <p>最近看了下网上的RxJava源码分析，发现所基于的源码版本和最新的略有不同，于是自己动手翻阅了一下最新的源码版本（rxjava:2.2.8，rxandroid:2.1.1），并写分析博客作分享。</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt"> 1
</span><span class="lnt"> 2
</span><span class="lnt"> 3
</span><span class="lnt"> 4
</span><span class="lnt"> 5
</span><span class="lnt"> 6
</span><span class="lnt"> 7
</span><span class="lnt"> 8
</span><span class="lnt"> 9
</span><span class="lnt">10
</span><span class="lnt">11
</span><span class="lnt">12
</span><span class="lnt">13
</span><span class="lnt">14
</span><span class="lnt">15
</span><span class="lnt">16
</span><span class="lnt">17
</span><span class="lnt">18
</span><span class="lnt">19
</span><span class="lnt">20
</span><span class="lnt">21
</span><span class="lnt">22
</span><span class="lnt">23
</span><span class="lnt">24
</span><span class="lnt">25
</span><span class="lnt">26
</span><span class="lnt">27
</span><span class="lnt">28
</span><span class="lnt">29
</span><span class="lnt">30
</span><span class="lnt">31
</span><span class="lnt">32
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="c1">//示例代码
</span><span class="c1"></span><span class="kd">private</span> <span class="kd">static</span> <span class="kt">void</span> <span class="nf">rxJavaTest</span><span class="o">()</span> <span class="o">{</span>
    <span class="n">Observable</span><span class="o">.</span><span class="na">create</span><span class="o">(</span><span class="k">new</span> <span class="n">ObservableOnSubscribe</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;()</span> <span class="o">{</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">subscribe</span><span class="o">(</span><span class="n">ObservableEmitter</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="n">emitter</span><span class="o">)</span> <span class="o">{</span>
              <span class="c1">//1
</span><span class="c1"></span>            <span class="n">emitter</span><span class="o">.</span><span class="na">onNext</span><span class="o">(</span><span class="s">&#34;onNext&#34;</span><span class="o">);</span>
            <span class="n">emitter</span><span class="o">.</span><span class="na">onComplete</span><span class="o">();</span>
        <span class="o">}</span>
    <span class="o">}).</span><span class="na">subscribe</span><span class="o">(</span><span class="k">new</span> <span class="n">Observer</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;()</span> <span class="o">{</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onSubscribe</span><span class="o">(</span><span class="n">Disposable</span> <span class="n">d</span><span class="o">)</span> <span class="o">{</span>
                <span class="c1">//2
</span><span class="c1"></span>               <span class="n">Log</span><span class="o">.</span><span class="na">d</span><span class="o">(</span><span class="n">TAG</span><span class="o">,</span> <span class="s">&#34;onSubscribe&#34;</span><span class="o">);</span>
        <span class="o">}</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onNext</span><span class="o">(</span><span class="n">String</span> <span class="n">s</span><span class="o">)</span> <span class="o">{</span>
              <span class="c1">//3
</span><span class="c1"></span>              <span class="n">Log</span><span class="o">.</span><span class="na">d</span><span class="o">(</span><span class="n">TAG</span><span class="o">,</span> <span class="n">s</span><span class="o">);</span>
        <span class="o">}</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onError</span><span class="o">(</span><span class="n">Throwable</span> <span class="n">e</span><span class="o">)</span> <span class="o">{</span>
              <span class="c1">//4
</span><span class="c1"></span>              <span class="n">Log</span><span class="o">.</span><span class="na">d</span><span class="o">(</span><span class="n">TAG</span><span class="o">,</span> <span class="s">&#34;onError&#34;</span><span class="o">);</span>
        <span class="o">}</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onComplete</span><span class="o">()</span> <span class="o">{</span>
              <span class="c1">//5
</span><span class="c1"></span>              <span class="n">Log</span><span class="o">.</span><span class="na">d</span><span class="o">(</span><span class="n">TAG</span><span class="o">,</span> <span class="s">&#34;onComplete&#34;</span><span class="o">);</span>
        <span class="o">}</span>
    <span class="o">});</span>
<span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>上面RxJava最简单的使用，主要涉及被观察者Observable、观察者Observer和事件订阅subscribe()三个角色。<br>
首先分析Observable的创建过程，即Observable的create()方法：</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt">1
</span><span class="lnt">2
</span><span class="lnt">3
</span><span class="lnt">4
</span><span class="lnt">5
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">static</span> <span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="n">Observable</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="nf">create</span><span class="o">(</span><span class="n">ObservableOnSubscribe</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="n">source</span><span class="o">)</span> <span class="o">{</span>
    <span class="n">ObjectHelper</span><span class="o">.</span><span class="na">requireNonNull</span><span class="o">(</span><span class="n">source</span><span class="o">,</span> <span class="s">&#34;source is null&#34;</span><span class="o">);</span>
    <span class="c1">//这里传入的source对象是我们传入的匿名内部类
</span><span class="c1"></span>    <span class="k">return</span> <span class="n">RxJavaPlugins</span><span class="o">.</span><span class="na">onAssembly</span><span class="o">(</span><span class="k">new</span> <span class="n">ObservableCreate</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;(</span><span class="n">source</span><span class="o">));</span>
<span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>这里先看一下我们传入的匿名内部类类型ObservableOnSubscribe源码：</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt">1
</span><span class="lnt">2
</span><span class="lnt">3
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">interface</span> <span class="nc">ObservableOnSubscribe</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="o">{</span>
    <span class="kt">void</span> <span class="nf">subscribe</span><span class="o">(</span><span class="nd">@NonNull</span> <span class="n">ObservableEmitter</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="n">emitter</span><span class="o">)</span> <span class="kd">throws</span> <span class="n">Exception</span><span class="o">;</span>
<span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>可见ObservableOnSubscribe是一个只含有一个抽象方法subscribe()的接口。</p>
<p>接着调用RxJavaPlugins的onAssembly()方法并传入一个新建的ObservableCreate对象，而ObservableCreate的构造函数如下：</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt">1
</span><span class="lnt">2
</span><span class="lnt">3
</span><span class="lnt">4
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="kd">final</span> <span class="n">ObservableOnSubscribe</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="n">source</span><span class="o">;</span>
<span class="kd">public</span> <span class="nf">ObservableCreate</span><span class="o">(</span><span class="n">ObservableOnSubscribe</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="n">source</span><span class="o">)</span> <span class="o">{</span>
   <span class="k">this</span><span class="o">.</span><span class="na">source</span> <span class="o">=</span> <span class="n">source</span><span class="o">;</span>
<span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>内部操作很简单，只是把ObservableCreate的成员变量source赋值为传入的ObservableOnSubscribe对象，即最开始我们创建的匿名内部类。<br>
RxJavaPlugins的onAssembly()方法调用如下：</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt">1
</span><span class="lnt">2
</span><span class="lnt">3
</span><span class="lnt">4
</span><span class="lnt">5
</span><span class="lnt">6
</span><span class="lnt">7
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">static</span> <span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="n">Observable</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="nf">onAssembly</span><span class="o">(</span><span class="nd">@NonNull</span> <span class="n">Observable</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="n">source</span><span class="o">)</span> <span class="o">{</span>
    <span class="n">Function</span><span class="o">&lt;?</span> <span class="kd">super</span> <span class="n">Observable</span><span class="o">,</span> <span class="o">?</span> <span class="kd">extends</span> <span class="n">Observable</span><span class="o">&gt;</span> <span class="n">f</span> <span class="o">=</span> <span class="n">onObservableAssembly</span><span class="o">;</span>
    <span class="k">if</span> <span class="o">(</span><span class="n">f</span> <span class="o">!=</span> <span class="kc">null</span><span class="o">)</span> <span class="o">{</span>
        <span class="k">return</span> <span class="n">apply</span><span class="o">(</span><span class="n">f</span><span class="o">,</span> <span class="n">source</span><span class="o">);</span>
    <span class="o">}</span>
    <span class="k">return</span> <span class="n">source</span><span class="o">;</span>
<span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>首先会创建一个Function f，赋值为onObservableAssembly，而onObservableAssembly默认为null（当使用转换操作符时会进行赋值，在后面的文章中会进一步分析），所以会直接返回source。至此Observable创建完毕。<br>
然后看Observer的内部实现，是一个包含4个抽象方法的接口：</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt">1
</span><span class="lnt">2
</span><span class="lnt">3
</span><span class="lnt">4
</span><span class="lnt">5
</span><span class="lnt">6
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">interface</span> <span class="nc">Observer</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="o">{</span>
    <span class="kt">void</span> <span class="nf">onSubscribe</span><span class="o">(</span><span class="nd">@NonNull</span> <span class="n">Disposable</span> <span class="n">d</span><span class="o">);</span>
    <span class="kt">void</span> <span class="nf">onNext</span><span class="o">(</span><span class="nd">@NonNull</span> <span class="n">T</span> <span class="n">t</span><span class="o">);</span>
    <span class="kt">void</span> <span class="nf">onError</span><span class="o">(</span><span class="nd">@NonNull</span> <span class="n">Throwable</span> <span class="n">e</span><span class="o">);</span>
    <span class="kt">void</span> <span class="nf">onComplete</span><span class="o">();</span>
<span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>最后来看重点subscribe()方法：</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt"> 1
</span><span class="lnt"> 2
</span><span class="lnt"> 3
</span><span class="lnt"> 4
</span><span class="lnt"> 5
</span><span class="lnt"> 6
</span><span class="lnt"> 7
</span><span class="lnt"> 8
</span><span class="lnt"> 9
</span><span class="lnt">10
</span><span class="lnt">11
</span><span class="lnt">12
</span><span class="lnt">13
</span><span class="lnt">14
</span><span class="lnt">15
</span><span class="lnt">16
</span><span class="lnt">17
</span><span class="lnt">18
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="nd">@Override</span>
<span class="kd">public</span> <span class="kd">final</span> <span class="kt">void</span> <span class="nf">subscribe</span><span class="o">(</span><span class="n">Observer</span><span class="o">&lt;?</span> <span class="kd">super</span> <span class="n">T</span><span class="o">&gt;</span> <span class="n">observer</span><span class="o">)</span> <span class="o">{</span>
    <span class="n">ObjectHelper</span><span class="o">.</span><span class="na">requireNonNull</span><span class="o">(</span><span class="n">observer</span><span class="o">,</span> <span class="s">&#34;observer is null&#34;</span><span class="o">);</span>
    <span class="k">try</span> <span class="o">{</span>
       
        <span class="n">observer</span> <span class="o">=</span> <span class="n">RxJavaPlugins</span><span class="o">.</span><span class="na">onSubscribe</span><span class="o">(</span><span class="k">this</span><span class="o">,</span> <span class="n">observer</span><span class="o">);</span><span class="c1">//1
</span><span class="c1"></span>        <span class="n">ObjectHelper</span><span class="o">.</span><span class="na">requireNonNull</span><span class="o">(</span><span class="n">observer</span><span class="o">,</span> <span class="s">&#34;The RxJavaPlugins.onSubscribe hook returned a null Observer...&#34;</span><span class="o">);</span>
        <span class="n">subscribeActual</span><span class="o">(</span><span class="n">observer</span><span class="o">);</span><span class="c1">//2
</span><span class="c1"></span>    <span class="o">}</span> <span class="k">catch</span> <span class="o">(</span><span class="n">NullPointerException</span> <span class="n">e</span><span class="o">)</span> <span class="o">{</span> 
        <span class="k">throw</span> <span class="n">e</span><span class="o">;</span>
    <span class="o">}</span> <span class="k">catch</span> <span class="o">(</span><span class="n">Throwable</span> <span class="n">e</span><span class="o">)</span> <span class="o">{</span>
        <span class="n">Exceptions</span><span class="o">.</span><span class="na">throwIfFatal</span><span class="o">(</span><span class="n">e</span><span class="o">);</span>
        <span class="n">RxJavaPlugins</span><span class="o">.</span><span class="na">onError</span><span class="o">(</span><span class="n">e</span><span class="o">);</span>
        <span class="n">NullPointerException</span> <span class="n">npe</span> <span class="o">=</span> <span class="k">new</span> <span class="n">NullPointerException</span><span class="o">(</span><span class="s">&#34;Actually not, but can&#39;t throw other exceptions due to RS&#34;</span><span class="o">);</span>
        <span class="n">npe</span><span class="o">.</span><span class="na">initCause</span><span class="o">(</span><span class="n">e</span><span class="o">);</span>
        <span class="k">throw</span> <span class="n">npe</span><span class="o">;</span>
    <span class="o">}</span>
<span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>在1处调用RxJavaPlugins.onSubscribe()，将我们传入的observer进行包装：</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt">1
</span><span class="lnt">2
</span><span class="lnt">3
</span><span class="lnt">4
</span><span class="lnt">5
</span><span class="lnt">6
</span><span class="lnt">7
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="kd">public</span> <span class="kd">static</span> <span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="n">Observer</span><span class="o">&lt;?</span> <span class="kd">super</span> <span class="n">T</span><span class="o">&gt;</span> <span class="nf">onSubscribe</span><span class="o">(</span><span class="nd">@NonNull</span> <span class="n">Observable</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="n">source</span><span class="o">,</span> <span class="nd">@NonNull</span> <span class="n">Observer</span><span class="o">&lt;?</span> <span class="kd">super</span> <span class="n">T</span><span class="o">&gt;</span> <span class="n">observer</span><span class="o">)</span> <span class="o">{</span>
    <span class="n">BiFunction</span><span class="o">&lt;?</span> <span class="kd">super</span> <span class="n">Observable</span><span class="o">,</span> <span class="o">?</span> <span class="kd">super</span> <span class="n">Observer</span><span class="o">,</span> <span class="o">?</span> <span class="kd">extends</span> <span class="n">Observer</span><span class="o">&gt;</span> <span class="n">f</span> <span class="o">=</span> <span class="n">onObservableSubscribe</span><span class="o">;</span>
    <span class="k">if</span> <span class="o">(</span><span class="n">f</span> <span class="o">!=</span> <span class="kc">null</span><span class="o">)</span> <span class="o">{</span>
        <span class="k">return</span> <span class="n">apply</span><span class="o">(</span><span class="n">f</span><span class="o">,</span> <span class="n">source</span><span class="o">,</span> <span class="n">observer</span><span class="o">);</span>
    <span class="o">}</span>
    <span class="k">return</span> <span class="n">observer</span><span class="o">;</span>
<span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>和上文的onAssembly()方法一样，这里的onObservableSubscribe也默认为null，所以返回值还是我们传入的observer本身。<br>
重中之重在于2处的subscribeActual(observer)：</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt"> 1
</span><span class="lnt"> 2
</span><span class="lnt"> 3
</span><span class="lnt"> 4
</span><span class="lnt"> 5
</span><span class="lnt"> 6
</span><span class="lnt"> 7
</span><span class="lnt"> 8
</span><span class="lnt"> 9
</span><span class="lnt">10
</span><span class="lnt">11
</span><span class="lnt">12
</span><span class="lnt">13
</span><span class="lnt">14
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="c1">//subscribeActual()是Observable中的抽象方法，本文示例的具体实现是在ObservableCreate类中
</span><span class="c1"></span><span class="kd">protected</span> <span class="kt">void</span> <span class="nf">subscribeActual</span><span class="o">(</span><span class="n">Observer</span><span class="o">&lt;?</span> <span class="kd">super</span> <span class="n">T</span><span class="o">&gt;</span> <span class="n">observer</span><span class="o">)</span> <span class="o">{</span>
    <span class="c1">//1、创建事件发射器
</span><span class="c1"></span>    <span class="n">CreateEmitter</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="n">parent</span> <span class="o">=</span> <span class="k">new</span> <span class="n">CreateEmitter</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;(</span><span class="n">observer</span><span class="o">);</span>
    <span class="c1">//2、调用observer的onSubscribe()方法
</span><span class="c1"></span>    <span class="n">observer</span><span class="o">.</span><span class="na">onSubscribe</span><span class="o">(</span><span class="n">parent</span><span class="o">);</span>
    <span class="k">try</span> <span class="o">{</span>
        <span class="c1">//3、调用source（即包装过的observer）的onSubscribe()方法，传入发射器
</span><span class="c1"></span>        <span class="n">source</span><span class="o">.</span><span class="na">subscribe</span><span class="o">(</span><span class="n">parent</span><span class="o">);</span>
    <span class="o">}</span> <span class="k">catch</span> <span class="o">(</span><span class="n">Throwable</span> <span class="n">ex</span><span class="o">)</span> <span class="o">{</span>
        <span class="n">Exceptions</span><span class="o">.</span><span class="na">throwIfFatal</span><span class="o">(</span><span class="n">ex</span><span class="o">);</span>
        <span class="n">parent</span><span class="o">.</span><span class="na">onError</span><span class="o">(</span><span class="n">ex</span><span class="o">);</span>
    <span class="o">}</span>
<span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>第1步，创建CreateEmitter，CreateEmitter类的主要代码如下：</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt"> 1
</span><span class="lnt"> 2
</span><span class="lnt"> 3
</span><span class="lnt"> 4
</span><span class="lnt"> 5
</span><span class="lnt"> 6
</span><span class="lnt"> 7
</span><span class="lnt"> 8
</span><span class="lnt"> 9
</span><span class="lnt">10
</span><span class="lnt">11
</span><span class="lnt">12
</span><span class="lnt">13
</span><span class="lnt">14
</span><span class="lnt">15
</span><span class="lnt">16
</span><span class="lnt">17
</span><span class="lnt">18
</span><span class="lnt">19
</span><span class="lnt">20
</span><span class="lnt">21
</span><span class="lnt">22
</span><span class="lnt">23
</span><span class="lnt">24
</span><span class="lnt">25
</span><span class="lnt">26
</span><span class="lnt">27
</span><span class="lnt">28
</span><span class="lnt">29
</span><span class="lnt">30
</span><span class="lnt">31
</span><span class="lnt">32
</span><span class="lnt">33
</span><span class="lnt">34
</span><span class="lnt">35
</span><span class="lnt">36
</span><span class="lnt">37
</span><span class="lnt">38
</span><span class="lnt">39
</span><span class="lnt">40
</span><span class="lnt">41
</span><span class="lnt">42
</span><span class="lnt">43
</span><span class="lnt">44
</span><span class="lnt">45
</span><span class="lnt">46
</span><span class="lnt">47
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="kd">static</span> <span class="kd">final</span> <span class="kd">class</span> <span class="nc">CreateEmitter</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;</span> <span class="kd">extends</span> <span class="n">AtomicReference</span><span class="o">&lt;</span><span class="n">Disposable</span><span class="o">&gt;</span>
    <span class="kd">implements</span> <span class="n">ObservableEmitter</span><span class="o">&lt;</span><span class="n">T</span><span class="o">&gt;,</span> <span class="n">Disposable</span> <span class="o">{</span>
        <span class="o">...</span>
        <span class="kd">final</span> <span class="n">Observer</span><span class="o">&lt;?</span> <span class="kd">super</span> <span class="n">T</span><span class="o">&gt;</span> <span class="n">observer</span><span class="o">;</span>
        <span class="n">CreateEmitter</span><span class="o">(</span><span class="n">Observer</span><span class="o">&lt;?</span> <span class="kd">super</span> <span class="n">T</span><span class="o">&gt;</span> <span class="n">observer</span><span class="o">)</span> <span class="o">{</span>
            <span class="c1">//将observer赋值给成员变量
</span><span class="c1"></span>            <span class="k">this</span><span class="o">.</span><span class="na">observer</span> <span class="o">=</span> <span class="n">observer</span><span class="o">;</span>
        <span class="o">}</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onNext</span><span class="o">(</span><span class="n">T</span> <span class="n">t</span><span class="o">)</span> <span class="o">{</span>
            <span class="k">if</span> <span class="o">(</span><span class="n">t</span> <span class="o">==</span> <span class="kc">null</span><span class="o">)</span> <span class="o">{</span>
                <span class="n">onError</span><span class="o">(</span><span class="k">new</span> <span class="n">NullPointerException</span><span class="o">(</span><span class="s">&#34;onNext called with null. Null values are generally not allowed in 2.x operators and sources.&#34;</span><span class="o">));</span>
                <span class="k">return</span><span class="o">;</span>
            <span class="o">}</span>
            <span class="k">if</span> <span class="o">(!</span><span class="n">isDisposed</span><span class="o">())</span> <span class="o">{</span>
                <span class="n">observer</span><span class="o">.</span><span class="na">onNext</span><span class="o">(</span><span class="n">t</span><span class="o">);</span>
            <span class="o">}</span>
        <span class="o">}</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onError</span><span class="o">(</span><span class="n">Throwable</span> <span class="n">t</span><span class="o">)</span> <span class="o">{</span>
            <span class="k">if</span> <span class="o">(!</span><span class="n">tryOnError</span><span class="o">(</span><span class="n">t</span><span class="o">))</span> <span class="o">{</span>
                <span class="n">RxJavaPlugins</span><span class="o">.</span><span class="na">onError</span><span class="o">(</span><span class="n">t</span><span class="o">);</span>
            <span class="o">}</span>
        <span class="o">}</span>
        <span class="o">...</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onComplete</span><span class="o">()</span> <span class="o">{</span>
            <span class="k">if</span> <span class="o">(!</span><span class="n">isDisposed</span><span class="o">())</span> <span class="o">{</span>
                <span class="k">try</span> <span class="o">{</span>
                    <span class="n">observer</span><span class="o">.</span><span class="na">onComplete</span><span class="o">();</span>
                <span class="o">}</span> <span class="k">finally</span> <span class="o">{</span>
                    <span class="n">dispose</span><span class="o">();</span>
                <span class="o">}</span>
            <span class="o">}</span>
        <span class="o">}</span>
        <span class="o">...</span>
        <span class="c1">//取消发送事件
</span><span class="c1"></span>        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">dispose</span><span class="o">()</span> <span class="o">{</span>
            <span class="n">DisposableHelper</span><span class="o">.</span><span class="na">dispose</span><span class="o">(</span><span class="k">this</span><span class="o">);</span>
        <span class="o">}</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">boolean</span> <span class="nf">isDisposed</span><span class="o">()</span> <span class="o">{</span>
            <span class="k">return</span> <span class="n">DisposableHelper</span><span class="o">.</span><span class="na">isDisposed</span><span class="o">(</span><span class="n">get</span><span class="o">());</span>
        <span class="o">}</span>
        <span class="o">...</span>
    <span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>第2步，调用observer的onSubscribe()方法，这样会走到示例代码的2处打印出&quot;onSubscribe&rdquo;。   <br>
第3步，调用source.subscribe(parent)，source实际上就是示例代码中的传入的匿名内部类：</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt">1
</span><span class="lnt">2
</span><span class="lnt">3
</span><span class="lnt">4
</span><span class="lnt">5
</span><span class="lnt">6
</span><span class="lnt">7
</span><span class="lnt">8
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="k">new</span> <span class="n">ObservableOnSubscribe</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;()</span> <span class="o">{</span>
   <span class="nd">@Override</span>
   <span class="kd">public</span> <span class="kt">void</span> <span class="nf">subscribe</span><span class="o">(</span><span class="n">ObservableEmitter</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;</span> <span class="n">emitter</span><span class="o">)</span> <span class="o">{</span>
       <span class="c1">//1
</span><span class="c1"></span>       <span class="n">emitter</span><span class="o">.</span><span class="na">onNext</span><span class="o">(</span><span class="s">&#34;onNext&#34;</span><span class="o">);</span>
       <span class="n">emitter</span><span class="o">.</span><span class="na">onComplete</span><span class="o">();</span>
   <span class="o">}</span>
<span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>所以会走到示例代码的1处分别执行emitter的onNext()和onComplete()方法，而从CreateEmitter的内部实现可见emitter的onNext()和onComplete()方法的具体操作实际上就是调用观察者observer的onNext()和onComplete()方法，observer即示例代码中的</p>
<div class="highlight"><div class="chroma">
<table class="lntable"><tr><td class="lntd">
<pre class="chroma"><code><span class="lnt"> 1
</span><span class="lnt"> 2
</span><span class="lnt"> 3
</span><span class="lnt"> 4
</span><span class="lnt"> 5
</span><span class="lnt"> 6
</span><span class="lnt"> 7
</span><span class="lnt"> 8
</span><span class="lnt"> 9
</span><span class="lnt">10
</span><span class="lnt">11
</span><span class="lnt">12
</span><span class="lnt">13
</span><span class="lnt">14
</span><span class="lnt">15
</span><span class="lnt">16
</span><span class="lnt">17
</span><span class="lnt">18
</span><span class="lnt">19
</span><span class="lnt">20
</span><span class="lnt">21
</span><span class="lnt">22
</span></code></pre></td>
<td class="lntd">
<pre class="chroma"><code class="language-java" data-lang="java"><span class="k">new</span> <span class="n">Observer</span><span class="o">&lt;</span><span class="n">String</span><span class="o">&gt;()</span> <span class="o">{</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onSubscribe</span><span class="o">(</span><span class="n">Disposable</span> <span class="n">d</span><span class="o">)</span> <span class="o">{</span>
                <span class="c1">//2
</span><span class="c1"></span>               <span class="n">Log</span><span class="o">.</span><span class="na">d</span><span class="o">(</span><span class="n">TAG</span><span class="o">,</span> <span class="s">&#34;onSubscribe&#34;</span><span class="o">);</span>
        <span class="o">}</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onNext</span><span class="o">(</span><span class="n">String</span> <span class="n">s</span><span class="o">)</span> <span class="o">{</span>
              <span class="c1">//3
</span><span class="c1"></span>              <span class="n">Log</span><span class="o">.</span><span class="na">d</span><span class="o">(</span><span class="n">TAG</span><span class="o">,</span> <span class="n">s</span><span class="o">);</span>
        <span class="o">}</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onError</span><span class="o">(</span><span class="n">Throwable</span> <span class="n">e</span><span class="o">)</span> <span class="o">{</span>
              <span class="c1">//4
</span><span class="c1"></span>              <span class="n">Log</span><span class="o">.</span><span class="na">d</span><span class="o">(</span><span class="n">TAG</span><span class="o">,</span> <span class="s">&#34;onError&#34;</span><span class="o">);</span>
        <span class="o">}</span>
        <span class="nd">@Override</span>
        <span class="kd">public</span> <span class="kt">void</span> <span class="nf">onComplete</span><span class="o">()</span> <span class="o">{</span>
              <span class="c1">//5
</span><span class="c1"></span>              <span class="n">Log</span><span class="o">.</span><span class="na">d</span><span class="o">(</span><span class="n">TAG</span><span class="o">,</span> <span class="s">&#34;onComplete&#34;</span><span class="o">);</span>
        <span class="o">}</span>
    <span class="o">}</span>
</code></pre></td></tr></table>
</div>
</div><p>于是就走到了示例代码3、4处，并在调用onNext()时传入1处emitter.onNext()方法传入的参数。<br>
以上只是RxJava最基础的用法的分析，主要是对观察者模式的不同角色进行封装，达到链式调用形式的目的，并且设计了发射器Emitter的概念，形成流式事件订阅的模式。</p>

    </div>

    <div class="post-copyright">
  <p class="copyright-item">
    <span class="item-title">文章作者</span>
    <span class="item-content">smartzheng</span>
  </p>
  <p class="copyright-item">
    <span class="item-title">上次更新</span>
    <span class="item-content">
        2019-05-05
        
    </span>
  </p>
  
  
</div>
<footer class="post-footer">
      
      <nav class="post-nav">
        <a class="prev" href="/post/navigation%E8%BF%94%E5%9B%9Efragment%E9%87%8D%E8%B5%B0oncreateview%E9%97%AE%E9%A2%98/">
            <i class="iconfont icon-left"></i>
            <span class="prev-text nav-default">Navigation返回Fragment重走onCreateView问题</span>
            <span class="prev-text nav-mobile">上一篇</span>
          </a>
        <a class="next" href="/post/android%E5%90%AF%E5%8A%A8%E4%BC%98%E5%8C%96%E6%9C%80%E4%BD%B3%E6%96%B9%E6%A1%88%E5%8E%BB%E5%90%AF%E5%8A%A8%E9%A1%B5%E5%92%8C%E5%BC%82%E6%AD%A5%E5%88%9D%E5%A7%8B%E5%8C%96/">
            <span class="next-text nav-default">Android启动优化最佳方案：去启动页和异步初始化</span>
            <span class="next-text nav-mobile">下一篇</span>
            <i class="iconfont icon-right"></i>
          </a>
      </nav>
    </footer>
  </article>
        </div>
        

  

  

      </div>
    </main>

    <footer id="footer" class="footer">
      <div class="social-links">
      <a href="mailto:smartzheng1993@email.com" class="iconfont icon-email" title="email"></a>
      <a href="https://github.com/smartzheng" class="iconfont icon-github" title="github"></a>
      <a href="https://weibo.com/2062336885/profile?topnav=1&amp;wvr=6" class="iconfont icon-weibo" title="weibo"></a>
      <a href="https://www.zhihu.com/people/liu-xia-hui-89-70" class="iconfont icon-zhihu" title="zhihu"></a>
      <a href="https://www.douban.com/people/smartzheng/" class="iconfont icon-douban" title="douban"></a>
  <a href="https://smartzheng.github.io/index.xml" type="application/rss+xml" class="iconfont icon-rss" title="rss"></a>
</div>

<div class="copyright">
  <span class="power-by">
    由 <a class="hexo-link" href="https://gohugo.io">Hugo</a> 强力驱动
  </span>
  <span class="division">|</span>
  <span class="theme-info">
    主题 - 
    <a class="theme-link" href="https://github.com/olOwOlo/hugo-theme-even">Even</a>
  </span>

  

  <span class="copyright-year">
    &copy; 
    2017 - 
    2020
    
  </span>
</div>
    </footer>

    <div class="back-to-top" id="back-to-top">
      <i class="iconfont icon-up"></i>
    </div>
  </div>
  
  <script src="https://cdn.jsdelivr.net/npm/jquery@3.2.1/dist/jquery.min.js" integrity="sha256-hwg4gsxgFZhOsEEamdOYGBf13FyQuiTwlAQgxVSNgt4=" crossorigin="anonymous"></script>
  <script src="https://cdn.jsdelivr.net/npm/slideout@1.0.1/dist/slideout.min.js" integrity="sha256-t+zJ/g8/KXIJMjSVQdnibt4dlaDxc9zXr/9oNPeWqdg=" crossorigin="anonymous"></script>
  <script src="https://cdn.jsdelivr.net/npm/@fancyapps/fancybox@3.1.20/dist/jquery.fancybox.min.js" integrity="sha256-XVLffZaxoWfGUEbdzuLi7pwaUJv1cecsQJQqGLe7axY=" crossorigin="anonymous"></script>
<script type="text/javascript" src="/dist/even.26188efa.min.js"></script>








</body>
</html>
